package com.spark

import com.KafkaPro
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming_SparkSQL {

  case class Record(classNum:String,sex:String)
  case class stuInfo(classNum:String,name:String,sex:String,birth:String,stuNum:String,term:String,gpa:String,state:String)
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("Test")
      .setMaster("local[2]")
    val ssc = new StreamingContext(conf,Seconds(30))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "node01:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> ("k5"+KafkaPro.GROUP_ID),
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (true: java.lang.Boolean)
    )

    val topics = Array("spark")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val mapDStream: DStream[stuInfo] = stream.filter(!_.value().isEmpty).map(_.value().split("\t")).map(s => stuInfo(s(0), s(1), s(2), s(3), s(4), s(5), s(6), s(7)))

    mapDStream.foreachRDD{rdd=>
      val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate()
      import spark.implicits._
      rdd.repartition(1).map(t => (t.classNum,t.sex))
        .toDF("classNum","sex").createOrReplaceTempView("temp")
      val frame: DataFrame = spark.sql("select classNum,sex,count(*) cnt from temp group by classNum,sex")
      frame.show()
      val url = "jdbc:mysql://mysql:3306/spark?serverTimezone=UTC"
      val table="part4"
      val prop = new java.util.Properties
      prop.setProperty("user","root")
      prop.setProperty("password","123456")
      frame.write.mode("append").jdbc(url, table, prop)
    }
    ssc.start()

    //等待计算结束
    ssc.awaitTermination()
  }

}